package com.shujia.spark.streaming

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Durations, StreamingContext}

object Demo3DStoRDD {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setMaster("local[2]").setAppName("wc")
    val sc = new SparkContext(conf)

    //创建sparkStraming上下文对象,  需要指定创建RDD的间隔时间    batch时间   可以理解为多少秒计算一次
    val ssc = new StreamingContext(sc, Durations.seconds(5))

    val ds = ssc.socketTextStream("node1", 8888)

    /**
      * ds  ---->   RDD
      *
      * DStream:每隔一段时间将接收过来的数据封装到一个rdd中进行计算， 所以DStream在api层面可以转换成rdd使用
      * rdd是不能转换成ds的，  流可以将其中一部分看作是微批，批不能看作是流
      *
      */

    //foreachRDD  将ds转换成rdd,使用rdd的api
    //没有返回值，里面的rdd必须使用action算子触发任务执行
    ds.foreachRDD(rdd => {

      println("属于Driver端的代码，每个batch会执行一次")

      //rdd.foreach(println)
    })


    /**
      * foreachRDD和transform都不会算子，运行在Driver端的方式，但是每一个batch都会运行一次
      */

    println("属于Driver端的代码，只会运行一次")

    /**
      *
      * transform：将ds转换成rdd,使用rdd的api,
      * 需要 返回一个新的rdd,在方法内部rdd可以不使用action算子
      * 闯入一个rdd返回一个rdd
      */
    val ds1 = ds.transform(rdd => {
      println("属于Driver端的代码，每个batch会执行一次")

      val countRDD = rdd.flatMap(_.split(",")).map((_, 1)).reduceByKey(_ + _)

      //返回新的偌大的，后面还可以使用ds的api
      countRDD
    })

    ds1.print()




    //启动sparkstreaming程序
    ssc.start()
    ssc.awaitTermination()
    ssc.stop()

  }

}
